package mysql

import (
	"database/sql"
	"fmt"
	"io"

	"github.com/example-user/go-pocket-etl/pkg/core"
	"github.com/example-user/go-pocket-etl/pkg/factory"
	// 匿名导入 MySQL 驱动，使其通过 init() 函数将自己注册到 database/sql 包中。
	// 这是 Go 中使用数据库驱动的标准方式。
	_ "github.com/go-sql-driver/mysql"
	"github.com/sirupsen/logrus"
)

// init 实现了组件的自注册。
func init() {
	factory.RegisterSource("mysql", NewSource)
}

// Source 实现了 core.Source 接口，用于从 MySQL 数据库读取数据。
type Source struct {
	db          *sql.DB   // 数据库连接池。它被设计为长期存活且线程安全。
	rows        *sql.Rows // SQL 查询结果集的前向只读迭代器。
	columnNames []string  // 预先获取的查询结果列名。
}

// NewSource 是 mysql.Source 的构造函数，由工厂调用。
func NewSource() core.Source {
	return &Source{}
}

// Open 负责解析配置、建立数据库连接并执行用户指定的查询。
// 它会初始化一个连接池，并准备好一个 `sql.Rows` 迭代器供 `Read` 方法消费。
func (s *Source) Open(config map[string]interface{}) error {
	// 'dsn' (Data Source Name) 是必需配置。
	dsn, ok := config["dsn"].(string)
	if !ok || dsn == "" {
		return fmt.Errorf("mysql source: config is missing or has invalid 'dsn'")
	}

	// 'query' 是必需配置。
	query, ok := config["query"].(string)
	if !ok || query == "" {
		return fmt.Errorf("mysql source: config is missing or has invalid 'query'")
	}

	var err error
	// sql.Open 并不会立即建立连接，而是返回一个代表连接池的 *sql.DB 对象。
	s.db, err = sql.Open("mysql", dsn)
	if err != nil {
		return fmt.Errorf("mysql source: failed to initialize database connection pool: %w", err)
	}

	// Ping 验证与数据库的连接是否真实有效，这是一个最佳实践。
	if err = s.db.Ping(); err != nil {
		return fmt.Errorf("mysql source: failed to connect to database: %w", err)
	}

	// 执行查询，获取结果集迭代器。
	logrus.Infof("Executing query: %s", query)
	s.rows, err = s.db.Query(query)
	if err != nil {
		return fmt.Errorf("mysql source: failed to execute query: %w", err)
	}

	// 预先获取并存储列名，这将在 Read 方法中用于构建 map[string]interface{} 格式的 Record。
	s.columnNames, err = s.rows.Columns()
	if err != nil {
		return fmt.Errorf("mysql source: failed to get column names from result set: %w", err)
	}

	return nil
}

// Read 读取查询结果的下一行，并将其转换为一个 `core.Record`。
func (s *Source) Read() (core.Record, error) {
	// 检查结果集中是否还有下一行。
	if !s.rows.Next() {
		// 在迭代结束后，必须调用 .Err() 来检查循环期间是否发生错误。
		if err := s.rows.Err(); err != nil {
			return nil, fmt.Errorf("mysql source: error during row iteration: %w", err)
		}
		// 如果没有错误，说明已成功到达结果集末尾，返回 EOF 信号。
		return nil, io.EOF
	}

	// 核心技巧：为了实现最大的通用性和健壮性，我们不直接扫描到具体的 Go 类型（如 int, string, time.Time），
	// 而是使用 `sql.RawBytes`。这有几个好处：
	// 1. 避免因数据库类型（如可为 NULL 的整数）与 Go 类型不匹配而导致的扫描错误。
	// 2. 优雅地处理 NULL 值（此时 RawBytes 切片为 nil）。
	// 3. 将所有值作为原始字节切片来处理，将类型转换的责任推迟到下游的 Processor，
	//    这使得 Source 组件更通用，更符合 ETL 的分阶段处理思想。
	values := make([]sql.RawBytes, len(s.columnNames))
	scanArgs := make([]interface{}, len(values))
	for i := range values {
		scanArgs[i] = &values[i]
	}

	// 将当前行的数据扫描到 `values` 中。
	if err := s.rows.Scan(scanArgs...); err != nil {
		return nil, fmt.Errorf("mysql source: failed to scan row: %w", err)
	}

	record := make(core.Record)
	for i, colName := range s.columnNames {
		// 如果数据库中的值是 NULL，对应的 RawBytes 切片将是 nil。
		if values[i] == nil {
			record[colName] = nil
		} else {
			// 否则，将其作为字符串存入 record。下游的 processor 可以根据需要再进行解析。
			record[colName] = string(values[i])
		}
	}

	return record, nil
}

// Close 负责优雅地关闭数据库资源。
// 它会先尝试关闭结果集迭代器，再关闭数据库连接池。
func (s *Source) Close() error {
	var firstErr error

	// 必须先关闭 rows。
	if s.rows != nil {
		if err := s.rows.Close(); err != nil {
			firstErr = fmt.Errorf("mysql source: failed to close rows: %w", err)
		}
	}
	// 然后关闭 db 连接池。
	if s.db != nil {
		if err := s.db.Close(); err != nil && firstErr == nil {
			firstErr = fmt.Errorf("mysql source: failed to close db connection: %w", err)
		}
	}

	return firstErr
}
